package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.model.common.constants.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
       return stream.flatMapValues(v->{
            UpdateArticleMess articleMess =JSON.parseObject(v, UpdateArticleMess.class);
            log.info("receive msg value: {}***************msg: {} ",v,articleMess);
            return Arrays.asList(articleMess.getArticleId()+":"+articleMess.getType());
        })
                .groupBy((k,v)->v)
                .windowedBy(TimeWindows.of(5000))
                .count()
                .toStream()
                .map((k,v)->{
                   log.info("After*********key: {},************value: {} ",k,v);
                   return new KeyValue<>(k.key().toString(),formatObj(k.key().toString(),v.toString()));
                });
    }


    private String formatObj(String k  , String v) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        String[] split = k.split(":");
        mess.setArticleId(Long.valueOf(split[0]));

        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
            mess.setLike( Long.valueOf(v) );
        }
        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
            mess.setView( Long.valueOf(v) );
        }
        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])) {
            mess.setComment( Long.valueOf(v) );
        }
        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
            mess.setCollect( Long.valueOf(v) );
        }

        log.info("formatObj aggs score ArticleVisitStreamMess:{}", mess);
        return JSON.toJSONString(mess);
    }

}